package com.atguigu.wordCount;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * @author gmd
 * @desc DataStream API实现WordCount（有界流）
 * @since 2024-11-23 05:47:13
 */
public class StreamOfReadFile {

    public static void main(String[] args) throws Exception {
        // 创建执行环境，流处理执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 读取数据，从文件读取
        DataStreamSource<String> lineDS = env.readTextFile("input/word.txt");

        // 3.处理数据: 切分、转换、分组、聚合
        // 3.1 切分、转换
        SingleOutputStreamOperator<Tuple2<String, Integer>> wordAndOneDS = lineDS
                .flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
                    @Override
                    public void flatMap(String value, Collector<Tuple2<String, Integer>> out) throws Exception {
                        // 按照 空格 切分
                        String[] words = value.split(" ");
                        for (String word : words) {
                            // 转换成 二元组 （word，1）
                            Tuple2<String, Integer> wordsAndOne = Tuple2.of(word, 1);
                            // 通过 采集器 向下游发送数据
                            out.collect(wordsAndOne);
                        }
                    }
                });

        // 3.2 分组
        KeyedStream<Tuple2<String, Integer>, String> wordAndOneKS = wordAndOneDS.keyBy(
                new KeySelector<Tuple2<String, Integer>, String>() {
                    @Override
                    public String getKey(Tuple2<String, Integer> value) throws Exception {
                        return value.f0;
                    }
                }
        );
        // 3.3 聚合
        SingleOutputStreamOperator<Tuple2<String, Integer>> sumDS = wordAndOneKS.sum(1);

        // 4.输出数据
        // 输出数据中，> 箭头左边的是任务槽ID
        sumDS.print();

        // 5.执行：类似 sparkstreaming最后 ssc.start()
        env.execute();
    }

    /*
     `BatchOfReadFile` 和 `StreamOfReadFile` 两个类都是实现 WordCount 的程序，但是它们使用了不同的 Flink API。
     `BatchOfReadFile` 使用的是 Flink 的 **DataSet API**，它是用于批处理的 API，适用于处理静态数据集。
     `StreamOfReadFile` 使用的是 Flink 的 **DataStream API**，它是用于流处理的 API，适用于处理实时数据流。
     简单来说，`BatchOfReadFile` 适用于处理文件中的静态数据，而 `StreamOfReadFile` 适用于处理实时数据流。
     在代码中，您可以看到 `BatchOfReadFile` 使用了 `ExecutionEnvironment` 和 `DataSource` 来读取文件数据，
     而 `StreamOfReadFile` 使用了 `StreamExecutionEnvironment` 和 `DataStream` 来处理实时数据流。
     */

}
